﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using PServiceBus.Core.Interface;
using System.Threading;
using PServiceBus.Core.Logger;
using PServiceBus.Core.Runtime;
using PServiceBus.Core.Manager;
using PServiceBus.Core.Runtime.Subscribers;
using PServiceBus.Core.Runtime.Extensions;
using PServiceBus.Core.Runtime.Transports;

namespace PServiceBus.SubscriberObservers {
    public class RedisSubscriberObserver : ISubscriberObserver {

        private ManualResetEvent _reset = new ManualResetEvent(false);
        private RegisteredWaitHandle _handle;
        private PServiceBus.Redis.Provider<ISubscriber> _provider = null;

        #region ISubscriberObserver Members

        public string Address { get; set; }

        public int Interval { get; set; }

        public void Stop() {
            _reset.Set();
        }

        private void ProcessEvent(object state, bool timeout) {
            if (timeout) {
                MethodHelper.TryLog(UpdateSubscribers);
                _handle = ThreadPool.RegisterWaitForSingleObject(_reset, new WaitOrTimerCallback(ProcessEvent),
                    null, Interval, true);
            } else {
                _handle.Unregister(null);
            }
        }

        private IObjectManager<ISubscriber> Subscribers {
            get { return ObjectManager<ISubscriber>.Instance; }
        }

        private void UpdateSubscribers() {
            var storedSubscribers = Subscribers.ToList();
            var subscribers = _provider.ToList();
            var addSubscribers = subscribers.Where(sub => !storedSubscribers.Any(s => sub.ID == s.ID));
            var deleteSubscribers = storedSubscribers.Where(sub => !subscribers.Any(s => sub.ID == s.ID));
            var currentSubscribers = subscribers.Where(sub => storedSubscribers.Any(s => sub.ID == s.ID));

            var subscriberCollection = SubscriberCollection.New;

            addSubscribers.Each(s => subscriberCollection.Append(s));

            currentSubscribers.Each(s => subscriberCollection.Refresh(s));

            subscriberCollection.Delete(deleteSubscribers.Select(s => s.ID));

            subscriberCollection.Save();
        }

        public void Init() {
            _provider = new PServiceBus.Redis.Provider<ISubscriber>(){ Endpoint = Address, AppendTypeNameToEndpoint = true, 
                IDFunc = s => s.ID};
        }

        public void Start() {
            _reset.Reset();
            ThreadPool.RegisterWaitForSingleObject(_reset, new WaitOrTimerCallback(ProcessEvent),
                null, Interval, true);
        }
        #endregion
    }
}
